package com.源码分析;

import java.util.ArrayList;
import java.util.Random;

import org.junit.Test;

import reactor.core.Fuseable;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class ThreadDemo3 {

	@Test
	// Flux的publishOn
	public void testSyncToAsync() throws InterruptedException {
		final Random random = new Random();
	    Flux.generate(ArrayList::new, (list, sink) -> {
	        int value = random.nextInt(100);
	        list.add(value);
	        System.out.println("线程" + Thread.currentThread().getName() + "发射元素");
	        sink.next(value);
	        try {
	            Thread.sleep(100);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        }
	        if (list.size() == 0) {
	            sink.complete();
	        }
	        return list;
	    })
	    .publishOn(Schedulers.parallel(), 2)
	    // 可以连着调用两次generate(...)，会导致publishOn(...)的行为有所不同，因为第一个的outputFused会设置为true，进而run()时会执行runBackfused()、第二个的sourceMode被设置为Fuseable.ASYNC
	    .publishOn(Schedulers.parallel(), 2)
	    .map(x -> String.format("[%s} %s", Thread.currentThread().getName(), x))
	    .subscribe(x -> {
	    	try {
	            Thread.sleep(300);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        }
	    	System.out.println(x);
	    });
	    Thread.sleep(3000000);
	}

}
